/*-------------------------------------------------------------------------
 * logical.c
 *	   PostgreSQL 逻辑解码协调
 *
 * Copyright (c) 2012-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  src/backend/replication/logical/logical.c
 *
 * NOTES
 *	  本文件协调提供逻辑解码的各个模块之间的交互，主要通过提供所谓的 LogicalDecodingContexts。目标是封装大部分逻辑解码的内部复杂性，以便其消费者可以用较少的代码创建和使用变更流。内置的消费者是 walsender 和 SQL SRF 接口，但可以在不更改核心代码的情况下添加更多消费者，例如在 bgworker 中消费变更。
 *
 *	  构思是消费者提供三个回调，一个用于读取 WAL，一个用于准备数据写入，最后一个用于实际写入，因为它们的实现取决于消费者的类型。请查看 logicalfuncs.c，以获取一个相对简单的消费者实现示例和一个适合简单消费者的 WAL 读取回调的实现。
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/xact.h"
#include "access/xlog_internal.h"
#include "fmgr.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "replication/decode.h"
#include "replication/logical.h"
#include "replication/origin.h"
#include "replication/reorderbuffer.h"
#include "replication/snapbuild.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "utils/builtins.h"
#include "utils/memutils.h"

/* errcontext 回调的数据 */
typedef struct LogicalErrorCallbackState
{
	LogicalDecodingContext *ctx;
	const char *callback_name;
	XLogRecPtr	report_location;
} LogicalErrorCallbackState;

/* 输出插件回调的包装器 */
static void fc_output_plugin_error_callback(void *fc_arg);
static void fc_startup_cb_wrapper(LogicalDecodingContext *fc_ctx, OutputPluginOptions *fc_opt,
							   bool fc_is_init);
static void fc_shutdown_cb_wrapper(LogicalDecodingContext *fc_ctx);
static void fc_begin_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn);
static void fc_commit_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
							  XLogRecPtr fc_commit_lsn);
static void fc_begin_prepare_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn);
static void fc_prepare_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
							   XLogRecPtr fc_prepare_lsn);
static void fc_commit_prepared_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
									   XLogRecPtr fc_commit_lsn);
static void fc_rollback_prepared_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
										 XLogRecPtr fc_prepare_end_lsn, TimestampTz fc_prepare_time);
static void fc_change_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
							  Relation fc_relation, ReorderBufferChange *fc_change);
static void fc_truncate_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
								int fc_nrelations, Relation fc_relations[], ReorderBufferChange *fc_change);
static void fc_message_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
							   XLogRecPtr fc_message_lsn, bool fc_transactional,
							   const char *fc_prefix, Size fc_message_size, const char *fc_message);

/* 流式回调 */
static void fc_stream_start_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
									XLogRecPtr fc_first_lsn);
static void fc_stream_stop_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
								   XLogRecPtr fc_last_lsn);
static void fc_stream_abort_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
									XLogRecPtr fc_abort_lsn);
static void fc_stream_prepare_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
									  XLogRecPtr fc_prepare_lsn);
static void fc_stream_commit_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
									 XLogRecPtr fc_commit_lsn);
static void fc_stream_change_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
									 Relation fc_relation, ReorderBufferChange *fc_change);
static void fc_stream_message_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
									  XLogRecPtr fc_message_lsn, bool fc_transactional,
									  const char *fc_prefix, Size fc_message_size, const char *fc_message);
static void fc_stream_truncate_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
									   int fc_nrelations, Relation fc_relations[], ReorderBufferChange *fc_change);

static void fc_LoadOutputPlugin(OutputPluginCallbacks *fc_callbacks, const char *fc_plugin);

/*
 * 确保当前设置和环境能够进行逻辑
 * 解码。
 */
void CheckLogicalDecodingRequirements(void)
{
	CheckSlotRequirements();

	/*
	 * 注意：添加新要求可能意味着 RestoreSlotFromDisk() 需要相同的检查。
	 */

	if (wal_level < WAL_LEVEL_LOGICAL)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("logical decoding requires wal_level >= logical")));

	if (MyDatabaseId == InvalidOid)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("logical decoding requires a database connection")));
#ifndef FDD
	/* ----
	 * TODO: 我们必须在不久的将来改变这一点... */
#else
	/* ----
	 * TODO: 允许从备用节点进行逻辑解码 */
#endif
	/*
	 * 基本上缺少三个东西来实现这一点：
	 * 1) 我们需要能够正确且快速地识别一个 LSN 属于哪个时间线
	 * 2) 我们需要始终强制启用 hot_standby_feedback，以便主服务器无法删除我们所需的行。
	 * 3) 支持删除引用数据库的复制槽，在 dbase_redo 中。由于 HS 恢复冲突，不能有任何活动的槽，因此这应该相对容易。
	 * ----
	 */
	if (RecoveryInProgress())
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("logical decoding cannot be used while in recovery")));
}

/*
 * CreateInitDecodingContext() 和 CreateDecodingContext() 的助手函数，
 * 执行常见任务。
 */
static LogicalDecodingContext *
fc_StartupDecodingContext(List *fc_output_plugin_options,
					   XLogRecPtr fc_start_lsn,
					   TransactionId fc_xmin_horizon,
					   bool fc_need_full_snapshot,
					   bool fc_fast_forward,
					   bool fc_in_create,
					   XLogReaderRoutine *fc_xl_routine,
					   LogicalOutputPluginWriterPrepareWrite fc_prepare_write,
					   LogicalOutputPluginWriterWrite fc_do_write,
					   LogicalOutputPluginWriterUpdateProgress fc_update_progress)
{
	ReplicationSlot *fc_slot;
	MemoryContext fc_context,
				fc_old_context;
	LogicalDecodingContext *fc_ctx;

	/* 更短的行... */
	fc_slot = MyReplicationSlot;

	fc_context = AllocSetContextCreate(CurrentMemoryContext,
									"Logical decoding context",
									ALLOCSET_DEFAULT_SIZES);
	fc_old_context = MemoryContextSwitchTo(fc_context);
	fc_ctx = palloc0(sizeof(LogicalDecodingContext));

	fc_ctx->context = fc_context;

	/*
	 * (重新)加载输出插件，以便我们现在能够检测到不良（已删除的）输出插件。
	 */
	if (!fc_fast_forward)
		fc_LoadOutputPlugin(&fc_ctx->callbacks, NameStr(fc_slot->data.plugin));

	/*
	 * 由于槽的 xmin 已经设置，我们可以宣布自己为一个
	 * 逻辑解码后端，在计算 xmin 地平线时不需要单独检查，
	 * 因为 xmin 是通过复制槽强制实施的。
	 *
	 * 我们只有在不处于事务中时才能这样做（即通过 walsender 流式传输更改时的情况），
	 * 否则，已经设置的快照/xid 将被忽略。这并不是一个特别令人烦恼的限制，
	 * 因为 SQL 接口无论如何无法用于流式传输。
	 */
	if (!IsTransactionOrTransactionBlock())
	{
		LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
		MyProc->statusFlags |= PROC_IN_LOGICAL_DECODING;
		ProcGlobal->statusFlags[MyProc->pgxactoff] = MyProc->statusFlags;
		LWLockRelease(ProcArrayLock);
	}

	fc_ctx->slot = fc_slot;

	fc_ctx->reader = XLogReaderAllocate(wal_segment_size, NULL, fc_xl_routine, fc_ctx);
	if (!fc_ctx->reader)
		ereport(ERROR,
				(errcode(ERRCODE_OUT_OF_MEMORY),
				 errmsg("out of memory"),
				 errdetail("Failed while allocating a WAL reading processor.")));

	fc_ctx->reorder = ReorderBufferAllocate();
	fc_ctx->snapshot_builder =
		AllocateSnapshotBuilder(fc_ctx->reorder, fc_xmin_horizon, fc_start_lsn,
								fc_need_full_snapshot, fc_slot->data.two_phase_at);

	fc_ctx->reorder->private_data = fc_ctx;

	/* 包装输出插件回调，以便我们可以添加错误上下文信息 */
	fc_ctx->reorder->begin = fc_begin_cb_wrapper;
	fc_ctx->reorder->apply_change = fc_change_cb_wrapper;
	fc_ctx->reorder->apply_truncate = fc_truncate_cb_wrapper;
	fc_ctx->reorder->commit = fc_commit_cb_wrapper;
	fc_ctx->reorder->message = fc_message_cb_wrapper;

	/*
	 * 为了支持流式传输，我们需要 start/stop/abort/commit/change
	 * 回调。消息和截断回调是可选的，类似于常规输出插件。
	 * 然而，我们在至少启用一种方法时启用流式传输，以便能够轻松识别缺失
	 * 的方法。
	 *
	 * 我们在这里决定它，但仅在包装器中稍后检查。
	 */
	fc_ctx->streaming = (fc_ctx->callbacks.stream_start_cb != NULL) ||
		(fc_ctx->callbacks.stream_stop_cb != NULL) ||
		(fc_ctx->callbacks.stream_abort_cb != NULL) ||
		(fc_ctx->callbacks.stream_commit_cb != NULL) ||
		(fc_ctx->callbacks.stream_change_cb != NULL) ||
		(fc_ctx->callbacks.stream_message_cb != NULL) ||
		(fc_ctx->callbacks.stream_truncate_cb != NULL);

	/*
	 * 流式回调
	 *
	 * stream_message 和 stream_truncate 回调是可选的，因此
	 * 当缺失时不会失败并给出 ERROR，包装器只是简单地什么都不做。
	 * 我们必须将 ReorderBuffer 回调设置为某个值，否则那里的调用将崩溃
	 * （我们不想将检查移动到那里）。
	 */
	fc_ctx->reorder->stream_start = fc_stream_start_cb_wrapper;
	fc_ctx->reorder->stream_stop = fc_stream_stop_cb_wrapper;
	fc_ctx->reorder->stream_abort = fc_stream_abort_cb_wrapper;
	fc_ctx->reorder->stream_prepare = fc_stream_prepare_cb_wrapper;
	fc_ctx->reorder->stream_commit = fc_stream_commit_cb_wrapper;
	fc_ctx->reorder->stream_change = fc_stream_change_cb_wrapper;
	fc_ctx->reorder->stream_message = fc_stream_message_cb_wrapper;
	fc_ctx->reorder->stream_truncate = fc_stream_truncate_cb_wrapper;


	/*
	 * 为了支持两阶段逻辑解码，我们需要
	 * begin_prepare/prepare/commit-prepare/abort-prepare 回调。
	 * filter_prepare 回调是可选的。尽管如此，当至少启用一种方法时，
	 * 我们启用两阶段逻辑解码，以便能够轻松识别缺失的方法。
	 *
	 * 我们在这里决定它，但仅在包装器中稍后检查。
	 */
	fc_ctx->twophase = (fc_ctx->callbacks.begin_prepare_cb != NULL) ||
		(fc_ctx->callbacks.prepare_cb != NULL) ||
		(fc_ctx->callbacks.commit_prepared_cb != NULL) ||
		(fc_ctx->callbacks.rollback_prepared_cb != NULL) ||
		(fc_ctx->callbacks.stream_prepare_cb != NULL) ||
		(fc_ctx->callbacks.filter_prepare_cb != NULL);

	/*
	 * 支持在准备时间解码的回调。
	 */
	fc_ctx->reorder->begin_prepare = fc_begin_prepare_cb_wrapper;
	fc_ctx->reorder->prepare = fc_prepare_cb_wrapper;
	fc_ctx->reorder->commit_prepared = fc_commit_prepared_cb_wrapper;
	fc_ctx->reorder->rollback_prepared = fc_rollback_prepared_cb_wrapper;

	fc_ctx->out = makeStringInfo();
	fc_ctx->prepare_write = fc_prepare_write;
	fc_ctx->write = fc_do_write;
	fc_ctx->update_progress = fc_update_progress;

	fc_ctx->output_plugin_options = fc_output_plugin_options;

	fc_ctx->fast_forward = fc_fast_forward;

	fc_ctx->in_create = fc_in_create;

	MemoryContextSwitchTo(fc_old_context);

	return fc_ctx;
}

/*
 * 为新的逻辑槽创建一个新的解码上下文。
 *
 * plugin -- 包含输出插件的名称
 * output_plugin_options -- 包含传递给输出插件的选项
 * need_full_snapshot -- 如果为 true，必须获取能够读取所有
 *		表的快照；如果为 false，能够仅读取目录的快照是可以接受的。
 * restart_lsn -- 如果作为无效提供，则该例程有责任
 *		通过设置一个方便的 restart_lsn 来标记 WAL 为保留。
 *		否则，我们设置从给定的 LSN 开始解码，而不事先
 *		标记 WAL 为保留。在这种情况下，由调用者
 *		保证 WAL 仍然可用。
 * xl_routine -- 适用于底层 XLogReader 的 XLogReaderRoutine
 * prepare_write, do_write, update_progress --
 *		执行用例相关实际工作的回调。
 *
 * 需要在一个内存上下文中调用，该上下文的生命周期至少与
 * 解码上下文一样长，因为将在其中创建进一步的内存上下文。
 *
 * 返回一个经过初始化的解码上下文，在调用输出插件的
 * 启动函数后。
 */
LogicalDecodingContext *
CreateInitDecodingContext(const char *fc_plugin,
						  List *fc_output_plugin_options,
						  bool fc_need_full_snapshot,
						  XLogRecPtr fc_restart_lsn,
						  XLogReaderRoutine *fc_xl_routine,
						  LogicalOutputPluginWriterPrepareWrite fc_prepare_write,
						  LogicalOutputPluginWriterWrite fc_do_write,
						  LogicalOutputPluginWriterUpdateProgress fc_update_progress)
{
	TransactionId fc_xmin_horizon = InvalidTransactionId;
	ReplicationSlot *fc_slot;
	NameData	fc_plugin_name;
	LogicalDecodingContext *fc_ctx;
	MemoryContext fc_old_context;

	/* 更短的行... */
	fc_slot = MyReplicationSlot;

	/* 首先进行一些不太可能被违反的完整性检查 */
	if (fc_slot == NULL)
		elog(ERROR, "cannot perform logical decoding without an acquired slot");

	if (fc_plugin == NULL)
		elog(ERROR, "cannot initialize logical decoding without a specified plugin");

	/* 确保传递的槽是合适的。这些是用户面临的错误。 */
	if (SlotIsPhysical(fc_slot))
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("cannot use physical replication slot for logical decoding")));

	if (fc_slot->data.database != MyDatabaseId)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("replication slot \"%s\" was not created in this database",
						NameStr(fc_slot->data.name))));

	if (IsTransactionState() &&
		GetTopTransactionIdIfAny() != InvalidTransactionId)
		ereport(ERROR,
				(errcode(ERRCODE_ACTIVE_SQL_TRANSACTION),
				 errmsg("cannot create logical replication slot in transaction that has performed writes")));

	/*
	 * 注册输出插件名称与槽。我们需要互斥锁以避免
	 * 同时读取部分复制的字符串。但我们不希望在持有自旋锁时进行任何复杂代码，
	 * 所以在外部执行 namestrcpy()。
	 */
	namestrcpy(&fc_plugin_name, fc_plugin);
	SpinLockAcquire(&fc_slot->mutex);
	fc_slot->data.plugin = fc_plugin_name;
	SpinLockRelease(&fc_slot->mutex);

	if (XLogRecPtrIsInvalid(fc_restart_lsn))
		ReplicationSlotReserveWal();
	else
	{
		SpinLockAcquire(&fc_slot->mutex);
		fc_slot->data.restart_lsn = fc_restart_lsn;
		SpinLockRelease(&fc_slot->mutex);
	}

	
/* ----
	 * 这有点棘手：我们需要确定一个安全的 xmin 范围以开始
	 * 解码，以避免从引用已经被清理或修剪的 xids 的正在运行的 xacts 记录开始。
	 * GetOldestSafeDecodingTransactionId() 返回这样的值，但
	 * 如果没有进一步的互锁，返回值可能会立即过时。
	 *
	 * 因此，我们必须获取 ProcArrayLock，以防止其他后端计算新的
	 * xmin 范围，获取安全的解码 xid，并告知
	 * 插槽机制新的限制。一旦完成，ProcArrayLock 可以被释放，因为插槽机制现在
	 * 正在保护免受清理。
	 *
	 * 请注意，暂时地，如果要导出数据快照，数据和目录
	 * 的 xmin 都必须被保留。否则，这里创建的初始
	 * 数据快照无法保证有效。在此之后，
	 * 数据 xmin 不需要再管理，全球 xmin 应该重新计算。由于我们在崩溃后没问题失去固定的数据 xmin - 不会有快照再被导出 - 我们可以
	 * 只需设置插槽的
	 * effective_xmin。ReplicationSlotRelease 会再次重置它。
	 *
	 * ----
	 */
	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);

	fc_xmin_horizon = GetOldestSafeDecodingTransactionId(!fc_need_full_snapshot);

	SpinLockAcquire(&fc_slot->mutex);
	fc_slot->effective_catalog_xmin = fc_xmin_horizon;
	fc_slot->data.catalog_xmin = fc_xmin_horizon;
	if (fc_need_full_snapshot)
		fc_slot->effective_xmin = fc_xmin_horizon;
	SpinLockRelease(&fc_slot->mutex);

	ReplicationSlotsComputeRequiredXmin(true);

	LWLockRelease(ProcArrayLock);

	ReplicationSlotMarkDirty();
	ReplicationSlotSave();

	fc_ctx = fc_StartupDecodingContext(NIL, fc_restart_lsn, fc_xmin_horizon,
								 fc_need_full_snapshot, false, true,
								 fc_xl_routine, fc_prepare_write, fc_do_write,
								 fc_update_progress);

	/* 调用输出插件初始化回调 */
	fc_old_context = MemoryContextSwitchTo(fc_ctx->context);
	if (fc_ctx->callbacks.startup_cb != NULL)
		fc_startup_cb_wrapper(fc_ctx, &fc_ctx->options, true);
	MemoryContextSwitchTo(fc_old_context);

	/*
	 * 当在插槽创建时启用 two_phase，或者在流开始时提供
	 * two_phase 选项时，我们允许解码已准备的事务，前提是插件支持所有
	 * 对于两阶段的回调。
	 */
	fc_ctx->twophase &= fc_slot->data.two_phase;

	fc_ctx->reorder->output_rewrites = fc_ctx->options.receive_rewrites;

	return fc_ctx;
}

/*
 * 创建一个新的解码上下文，用于先前已使用的逻辑插槽。
 *
 * start_lsn
 *		开始解码的 LSN。如果是 InvalidXLogRecPtr，从插槽的 confirmed_flush 重新启动；否则，从指定
 *		的位置开始（但如果它比 confirmed_flush 旧，则向前移动到 confirmed_flush）。
 *
 * output_plugin_options
 *		传递给输出插件的选项。
 *
 * fast_forward
 *		绕过逻辑更改的生成。
 *
 * xl_routine
 *		由底层 xlogreader 使用的 XLogReaderRoutine。
 *
 * prepare_write, do_write, update_progress
 *		必须填写的回调，以执行用例相关的实际工作。
 *
 * 需要在一个至少与解码上下文同样长寿命的内存上下文中调用，因为会在
 * 其中创建进一步的内存上下文。
 *
 * 在调用输出插件的启动函数后返回初始化的解码上下文。
 */
LogicalDecodingContext *
CreateDecodingContext(XLogRecPtr fc_start_lsn,
					  List *fc_output_plugin_options,
					  bool fc_fast_forward,
					  XLogReaderRoutine *fc_xl_routine,
					  LogicalOutputPluginWriterPrepareWrite fc_prepare_write,
					  LogicalOutputPluginWriterWrite fc_do_write,
					  LogicalOutputPluginWriterUpdateProgress fc_update_progress)
{
	LogicalDecodingContext *fc_ctx;
	ReplicationSlot *fc_slot;
	MemoryContext fc_old_context;

	/* 更短的行... */
	fc_slot = MyReplicationSlot;

	/* 首先进行一些不太可能被违反的完整性检查 */
	if (fc_slot == NULL)
		elog(ERROR, "cannot perform logical decoding without an acquired slot");

	/* 确保传递的插槽是合适的，这些都是用户相关的错误 */
	if (SlotIsPhysical(fc_slot))
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("cannot use physical replication slot for logical decoding")));

	if (fc_slot->data.database != MyDatabaseId)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("replication slot \"%s\" was not created in this database",
						NameStr(fc_slot->data.name))));

	if (fc_start_lsn == InvalidXLogRecPtr)
	{
		/* 从上一个位置继续 */
		fc_start_lsn = fc_slot->data.confirmed_flush;
	}
	else if (fc_start_lsn < fc_slot->data.confirmed_flush)
	{
		/*
		 * 在这种情况下，似乎我们应该报错，但客户端对一个它不需要
		 * 做任何事情的 LSN 的确认相当常见，因此没有持久存储，因为
		 * xlog 记录没有导致逻辑解码的任何相关结果。客户端必须能够这样做以支持同步
		 * 复制。
		 *
		 * 从请求的不同 LSN 开始可能无法捕捉某些类型的客户端错误；所以客户端可能希望检查
		 * confirmed_flush_lsn 是否与其期望匹配。
		 */
		elog(LOG, "%X/%X has been already streamed, forwarding to %X/%X",
			 LSN_FORMAT_ARGS(fc_start_lsn),
			 LSN_FORMAT_ARGS(fc_slot->data.confirmed_flush));

		fc_start_lsn = fc_slot->data.confirmed_flush;
	}

	fc_ctx = fc_StartupDecodingContext(fc_output_plugin_options,
								 fc_start_lsn, InvalidTransactionId, false,
								 fc_fast_forward, false, fc_xl_routine, fc_prepare_write,
								 fc_do_write, fc_update_progress);

	/* 调用输出插件初始化回调 */
	fc_old_context = MemoryContextSwitchTo(fc_ctx->context);
	if (fc_ctx->callbacks.startup_cb != NULL)
		fc_startup_cb_wrapper(fc_ctx, &fc_ctx->options, false);
	MemoryContextSwitchTo(fc_old_context);

	/*
	 * 当在插槽创建时启用 two_phase，或者在流开始时提供
	 * two_phase 选项时，我们允许解码已准备的事务，前提是插件支持所有
	 * 对于两阶段的回调。
	 */
	fc_ctx->twophase &= (fc_slot->data.two_phase || fc_ctx->twophase_opt_given);

	/* 标记插槽以允许 two_phase 解码（如果尚未标记） */
	if (fc_ctx->twophase && !fc_slot->data.two_phase)
	{
		SpinLockAcquire(&fc_slot->mutex);
		fc_slot->data.two_phase = true;
		fc_slot->data.two_phase_at = fc_start_lsn;
		SpinLockRelease(&fc_slot->mutex);
		ReplicationSlotMarkDirty();
		ReplicationSlotSave();
		SnapBuildSetTwoPhaseAt(fc_ctx->snapshot_builder, fc_start_lsn);
	}

	fc_ctx->reorder->output_rewrites = fc_ctx->options.receive_rewrites;

	ereport(LOG,
			(errmsg("starting logical decoding for slot \"%s\"",
					NameStr(fc_slot->data.name)),
			 errdetail("Streaming transactions committing after %X/%X, reading WAL from %X/%X.",
					   LSN_FORMAT_ARGS(fc_slot->data.confirmed_flush),
					   LSN_FORMAT_ARGS(fc_slot->data.restart_lsn))));

	return fc_ctx;
}

/*
 * 如果已构建一致的初始解码快照，则返回真。
 */
bool DecodingContextReady(LogicalDecodingContext *fc_ctx)
{
	return SnapBuildCurrentState(fc_ctx->snapshot_builder) == SNAPBUILD_CONSISTENT;
}

/*
 * 从解码插槽中读取，直到准备好开始提取更改。
 */
void DecodingContextFindStartpoint(LogicalDecodingContext *fc_ctx)
{
	ReplicationSlot *fc_slot = fc_ctx->slot;

	/* 初始化从哪里开始读取 WAL。 */
	XLogBeginRead(fc_ctx->reader, fc_slot->data.restart_lsn);

	elog(DEBUG1, "searching for logical decoding starting point, starting at %X/%X",
		 LSN_FORMAT_ARGS(fc_slot->data.restart_lsn));

	/* 等待一致的起始点 */
	for (;;)
	{
		XLogRecord *fc_record;
		char	   *fc_err = NULL;

		/* read_page 回调等待新的 WAL */
		fc_record = XLogReadRecord(fc_ctx->reader, &fc_err);
		if (fc_err)
			elog(ERROR, "could not find logical decoding starting point: %s", fc_err);
		if (!fc_record)
			elog(ERROR, "could not find logical decoding starting point");

		LogicalDecodingProcessRecord(fc_ctx, fc_ctx->reader);

		/* 仅继续到我们找到一个一致的位置 */
		if (DecodingContextReady(fc_ctx))
			break;

		CHECK_FOR_INTERRUPTS();
	}

	SpinLockAcquire(&fc_slot->mutex);
	fc_slot->data.confirmed_flush = fc_ctx->reader->EndRecPtr;
	if (fc_slot->data.two_phase)
		fc_slot->data.two_phase_at = fc_ctx->reader->EndRecPtr;
	SpinLockRelease(&fc_slot->mutex);
#ifdef FDD
	ReplicationSlotMarkDirty();
#endif

}

/*
 * 释放之前分配的解码上下文，如有必要调用关闭回调。
 */
void FreeDecodingContext(LogicalDecodingContext *fc_ctx)
{
	if (fc_ctx->callbacks.shutdown_cb != NULL)
		fc_shutdown_cb_wrapper(fc_ctx);

	ReorderBufferFree(fc_ctx->reorder);
	FreeSnapshotBuilder(fc_ctx->snapshot_builder);
	XLogReaderFree(fc_ctx->reader);
	MemoryContextDelete(fc_ctx->context);
}

/*
 * 使用上下文的输出例程准备写操作。
 */
void OutputPluginPrepareWrite(struct LogicalDecodingContext *fc_ctx, bool fc_last_write)
{
	if (!fc_ctx->accept_writes)
		elog(ERROR, "writes are only accepted in commit, begin and change callbacks");

	fc_ctx->prepare_write(fc_ctx, fc_ctx->write_location, fc_ctx->write_xid, fc_last_write);
	fc_ctx->prepared_write = true;
}

/*
 * 使用上下文的输出例程进行写操作。
 */
void OutputPluginWrite(struct LogicalDecodingContext *fc_ctx, bool fc_last_write)
{
	if (!fc_ctx->prepared_write)
		elog(ERROR, "OutputPluginPrepareWrite needs to be called before OutputPluginWrite");

	fc_ctx->write(fc_ctx, fc_ctx->write_location, fc_ctx->write_xid, fc_last_write);
	fc_ctx->prepared_write = false;
}

/*
 * 更新进度跟踪（如果支持的话）。
 */
void OutputPluginUpdateProgress(struct LogicalDecodingContext *fc_ctx,
						   bool fc_skipped_xact)
{
	if (!fc_ctx->update_progress)
		return;

	fc_ctx->update_progress(fc_ctx, fc_ctx->write_location, fc_ctx->write_xid,
						 fc_skipped_xact);
}

/*
 * 加载输出插件，查找其输出插件初始化函数，并检查
 * 它是否提供所需的回调。
 */
static void fc_LoadOutputPlugin(OutputPluginCallbacks *fc_callbacks, const char *fc_plugin)
{
	LogicalOutputPluginInit fc_plugin_init;

	fc_plugin_init = (LogicalOutputPluginInit)
		load_external_function(fc_plugin, "_PG_output_plugin_init", false, NULL);

	if (fc_plugin_init == NULL)
		elog(ERROR, "output plugins have to declare the _PG_output_plugin_init symbol");

	/* 请求输出插件填充回调结构 */
	fc_plugin_init(fc_callbacks);

	if (fc_callbacks->begin_cb == NULL)
		elog(ERROR, "output plugins have to register a begin callback");
	if (fc_callbacks->change_cb == NULL)
		elog(ERROR, "output plugins have to register a change callback");
	if (fc_callbacks->commit_cb == NULL)
		elog(ERROR, "output plugins have to register a commit callback");
}

static void fc_output_plugin_error_callback(void *fc_arg)
{
	LogicalErrorCallbackState *fc_state = (LogicalErrorCallbackState *) fc_arg;

	/* 并非所有回调都有相关的LSN */
	if (fc_state->report_location != InvalidXLogRecPtr)
		errcontext("slot \"%s\", output plugin \"%s\", in the %s callback, associated LSN %X/%X",
				   NameStr(fc_state->ctx->slot->data.name),
				   NameStr(fc_state->ctx->slot->data.plugin),
				   fc_state->callback_name,
				   LSN_FORMAT_ARGS(fc_state->report_location));
	else
		errcontext("slot \"%s\", output plugin \"%s\", in the %s callback",
				   NameStr(fc_state->ctx->slot->data.name),
				   NameStr(fc_state->ctx->slot->data.plugin),
				   fc_state->callback_name);
}

static void fc_startup_cb_wrapper(LogicalDecodingContext *fc_ctx, OutputPluginOptions *fc_opt, bool fc_is_init)
{
	LogicalErrorCallbackState fc_state;
	ErrorContextCallback fc_errcallback;

	Assert(!fc_ctx->fast_forward);

	/* 将回调 + 错误上下文信息推入栈中 */
	fc_state.ctx = fc_ctx;
	fc_state.callback_name = "startup";
	fc_state.report_location = InvalidXLogRecPtr;
	fc_errcallback.callback = fc_output_plugin_error_callback;
	fc_errcallback.arg = (void *) &fc_state;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 设置输出状态 */
	fc_ctx->accept_writes = false;
	fc_ctx->end_xact = false;

	/* 执行实际工作：调用回调 */
	fc_ctx->callbacks.startup_cb(fc_ctx, fc_opt, fc_is_init);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;
}

static void fc_shutdown_cb_wrapper(LogicalDecodingContext *fc_ctx)
{
	LogicalErrorCallbackState fc_state;
	ErrorContextCallback fc_errcallback;

	Assert(!fc_ctx->fast_forward);

	/* 将回调 + 错误上下文信息推入栈中 */
	fc_state.ctx = fc_ctx;
	fc_state.callback_name = "shutdown";
	fc_state.report_location = InvalidXLogRecPtr;
	fc_errcallback.callback = fc_output_plugin_error_callback;
	fc_errcallback.arg = (void *) &fc_state;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 设置输出状态 */
	fc_ctx->accept_writes = false;
	fc_ctx->end_xact = false;

	/* 执行实际工作：调用回调 */
	fc_ctx->callbacks.shutdown_cb(fc_ctx);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;
}


/*
 * ReorderBuffer的回调会添加一些额外信息，然后调用
 * output_plugin.h 插件。
 */
static void fc_begin_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn)
{
	LogicalDecodingContext *fc_ctx = fc_cache->private_data;
	LogicalErrorCallbackState fc_state;
	ErrorContextCallback fc_errcallback;

	Assert(!fc_ctx->fast_forward);

	/* 将回调 + 错误上下文信息推入栈中 */
	fc_state.ctx = fc_ctx;
	fc_state.callback_name = "begin";
	fc_state.report_location = fc_txn->first_lsn;
	fc_errcallback.callback = fc_output_plugin_error_callback;
	fc_errcallback.arg = (void *) &fc_state;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 设置输出状态 */
	fc_ctx->accept_writes = true;
	fc_ctx->write_xid = fc_txn->xid;
	fc_ctx->write_location = fc_txn->first_lsn;
	fc_ctx->end_xact = false;

	/* 执行实际工作：调用回调 */
	fc_ctx->callbacks.begin_cb(fc_ctx, fc_txn);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;
}

static void fc_commit_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
				  XLogRecPtr fc_commit_lsn)
{
	LogicalDecodingContext *fc_ctx = fc_cache->private_data;
	LogicalErrorCallbackState fc_state;
	ErrorContextCallback fc_errcallback;

	Assert(!fc_ctx->fast_forward);

	/* 将回调 + 错误上下文信息推入栈中 */
	fc_state.ctx = fc_ctx;
	fc_state.callback_name = "commit";
	fc_state.report_location = fc_txn->final_lsn; /* 提交记录的开始 */
	fc_errcallback.callback = fc_output_plugin_error_callback;
	fc_errcallback.arg = (void *) &fc_state;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 设置输出状态 */
	fc_ctx->accept_writes = true;
	fc_ctx->write_xid = fc_txn->xid;
	fc_ctx->write_location = fc_txn->end_lsn; /* 指向记录的结束 */
	fc_ctx->end_xact = true;

	/* 执行实际工作：调用回调 */
	fc_ctx->callbacks.commit_cb(fc_ctx, fc_txn, fc_commit_lsn);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;
}

/*
 * begin_prepare的功能与begin相类似，唯一的区别是
 * 这个将包含gid（全局事务ID）信息，可以被插件使用。
 * 现在，我们考虑过扩展现有的begin，但那样会破坏复制协议，此外这看起来更清晰。
 */
static void fc_begin_prepare_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn)
{
	LogicalDecodingContext *fc_ctx = fc_cache->private_data;
	LogicalErrorCallbackState fc_state;
	ErrorContextCallback fc_errcallback;

	Assert(!fc_ctx->fast_forward);

	/* 我们只应该在支持两阶段提交时调用此函数 */
	Assert(fc_ctx->twophase);

	/* 将回调 + 错误上下文信息推入栈中 */
	fc_state.ctx = fc_ctx;
	fc_state.callback_name = "begin_prepare";
	fc_state.report_location = fc_txn->first_lsn;
	fc_errcallback.callback = fc_output_plugin_error_callback;
	fc_errcallback.arg = (void *) &fc_state;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 设置输出状态 */
	fc_ctx->accept_writes = true;
	fc_ctx->write_xid = fc_txn->xid;
	fc_ctx->write_location = fc_txn->first_lsn;
	fc_ctx->end_xact = false;

	/*
	 * 如果插件支持两阶段提交，则begin prepare回调是
	 * 强制性的
	 */
	if (fc_ctx->callbacks.begin_prepare_cb == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("logical replication at prepare time requires a %s callback",
						"begin_prepare_cb")));

	/* 执行实际工作：调用回调 */
	fc_ctx->callbacks.begin_prepare_cb(fc_ctx, fc_txn);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;
}

static void fc_prepare_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
				   XLogRecPtr fc_prepare_lsn)
{
	LogicalDecodingContext *fc_ctx = fc_cache->private_data;
	LogicalErrorCallbackState fc_state;
	ErrorContextCallback fc_errcallback;

	Assert(!fc_ctx->fast_forward);

	/* 我们只应该在支持两阶段提交时调用此函数 */
	Assert(fc_ctx->twophase);

	/* 将回调 + 错误上下文信息推入栈中 */
	fc_state.ctx = fc_ctx;
	fc_state.callback_name = "prepare";
	fc_state.report_location = fc_txn->final_lsn; /* prepare记录的开始 */
	fc_errcallback.callback = fc_output_plugin_error_callback;
	fc_errcallback.arg = (void *) &fc_state;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 设置输出状态 */
	fc_ctx->accept_writes = true;
	fc_ctx->write_xid = fc_txn->xid;
	fc_ctx->write_location = fc_txn->end_lsn; /* 指向记录的结束 */
	fc_ctx->end_xact = true;

	/*
	 * 如果插件支持两阶段提交，则prepare回调是
	 * 强制性的
	 */
	if (fc_ctx->callbacks.prepare_cb == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("logical replication at prepare time requires a %s callback",
						"prepare_cb")));

	/* 执行实际工作：调用回调 */
	fc_ctx->callbacks.prepare_cb(fc_ctx, fc_txn, fc_prepare_lsn);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;
}

static void fc_commit_prepared_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
						   XLogRecPtr fc_commit_lsn)
{
	LogicalDecodingContext *fc_ctx = fc_cache->private_data;
	LogicalErrorCallbackState fc_state;
	ErrorContextCallback fc_errcallback;

	Assert(!fc_ctx->fast_forward);

	/* 我们只应该在支持两阶段提交时调用此函数 */
	Assert(fc_ctx->twophase);

	/* 将回调 + 错误上下文信息推入栈中 */
	fc_state.ctx = fc_ctx;
	fc_state.callback_name = "commit_prepared";
	fc_state.report_location = fc_txn->final_lsn; /* 提交记录的开始 */
	fc_errcallback.callback = fc_output_plugin_error_callback;
	fc_errcallback.arg = (void *) &fc_state;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 设置输出状态 */
	fc_ctx->accept_writes = true;
	fc_ctx->write_xid = fc_txn->xid;
	fc_ctx->write_location = fc_txn->end_lsn; /* 指向记录的结束 */
	fc_ctx->end_xact = true;

	/*
	 * 如果插件支持两阶段提交，则commit prepared回调
	 * 是强制性的
	 */
	if (fc_ctx->callbacks.commit_prepared_cb == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("logical replication at prepare time requires a %s callback",
						"commit_prepared_cb")));

	/* 执行实际工作：调用回调 */
	fc_ctx->callbacks.commit_prepared_cb(fc_ctx, fc_txn, fc_commit_lsn);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;
}

static void fc_rollback_prepared_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
							 XLogRecPtr fc_prepare_end_lsn,
							 TimestampTz fc_prepare_time)
{
	LogicalDecodingContext *fc_ctx = fc_cache->private_data;
	LogicalErrorCallbackState fc_state;
	ErrorContextCallback fc_errcallback;

	Assert(!fc_ctx->fast_forward);

	/* 我们只应该在支持两阶段提交时调用此函数 */
	Assert(fc_ctx->twophase);

	/* 将回调 + 错误上下文信息推入栈中 */
	fc_state.ctx = fc_ctx;
	fc_state.callback_name = "rollback_prepared";
	fc_state.report_location = fc_txn->final_lsn; /* 提交记录的开始 */
	fc_errcallback.callback = fc_output_plugin_error_callback;
	fc_errcallback.arg = (void *) &fc_state;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 设置输出状态 */
	fc_ctx->accept_writes = true;
	fc_ctx->write_xid = fc_txn->xid;
	fc_ctx->write_location = fc_txn->end_lsn; /* 指向记录的结束 */
	fc_ctx->end_xact = true;

	/*
	 * 如果插件支持两阶段提交，则rollback prepared回调
	 * 是强制性的
	 */
	if (fc_ctx->callbacks.rollback_prepared_cb == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("logical replication at prepare time requires a %s callback",
						"rollback_prepared_cb")));

	/* 执行实际工作：调用回调 */
	fc_ctx->callbacks.rollback_prepared_cb(fc_ctx, fc_txn, fc_prepare_end_lsn,
										fc_prepare_time);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;
}

static void fc_change_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
				  Relation fc_relation, ReorderBufferChange *fc_change)
{
	LogicalDecodingContext *fc_ctx = fc_cache->private_data;
	LogicalErrorCallbackState fc_state;
	ErrorContextCallback fc_errcallback;

	Assert(!fc_ctx->fast_forward);

	/* 将回调 + 错误上下文信息推入栈中 */
	fc_state.ctx = fc_ctx;
	fc_state.callback_name = "change";
	fc_state.report_location = fc_change->lsn;
	fc_errcallback.callback = fc_output_plugin_error_callback;
	fc_errcallback.arg = (void *) &fc_state;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 设置输出状态 */
	fc_ctx->accept_writes = true;
	fc_ctx->write_xid = fc_txn->xid;

	/*
	 * 报告此变更的LSN，以便客户端的回复可以给出最新的
	 * 答复。这永远不会足够（也不应该！）来确认
	 * 收到此事务，但它可能允许确认另一个事务的
	 * 提交只需一条消息。
	 */
	fc_ctx->write_location = fc_change->lsn;

	fc_ctx->end_xact = false;

	fc_ctx->callbacks.change_cb(fc_ctx, fc_txn, fc_relation, fc_change);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;
}

static void fc_truncate_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
					int fc_nrelations, Relation fc_relations[], ReorderBufferChange *fc_change)
{
	LogicalDecodingContext *fc_ctx = fc_cache->private_data;
	LogicalErrorCallbackState fc_state;
	ErrorContextCallback fc_errcallback;

	Assert(!fc_ctx->fast_forward);

	if (!fc_ctx->callbacks.truncate_cb)
		return;

	/* 将回调 + 错误上下文信息推入栈中 */
	fc_state.ctx = fc_ctx;
	fc_state.callback_name = "truncate";
	fc_state.report_location = fc_change->lsn;
	fc_errcallback.callback = fc_output_plugin_error_callback;
	fc_errcallback.arg = (void *) &fc_state;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 设置输出状态 */
	fc_ctx->accept_writes = true;
	fc_ctx->write_xid = fc_txn->xid;

	/*
	 * 报告此变更的LSN，以便客户端的回复可以给出最新的
	 * 答复。这永远不会足够（也不应该！）来确认
	 * 收到此事务，但它可能允许确认另一个事务的
	 * 提交只需一条消息。
	 */
	fc_ctx->write_location = fc_change->lsn;

	fc_ctx->end_xact = false;

	fc_ctx->callbacks.truncate_cb(fc_ctx, fc_txn, fc_nrelations, fc_relations, fc_change);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;
}

bool filter_prepare_cb_wrapper(LogicalDecodingContext *fc_ctx, TransactionId fc_xid,
						  const char *fc_gid)
{
	LogicalErrorCallbackState fc_state;
	ErrorContextCallback fc_errcallback;
	bool		fc_ret;

	Assert(!fc_ctx->fast_forward);

	/* 将回调 + 错误上下文信息推入栈中 */
	fc_state.ctx = fc_ctx;
	fc_state.callback_name = "filter_prepare";
	fc_state.report_location = InvalidXLogRecPtr;
	fc_errcallback.callback = fc_output_plugin_error_callback;
	fc_errcallback.arg = (void *) &fc_state;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 设置输出状态 */
	fc_ctx->accept_writes = false;
	fc_ctx->end_xact = false;

	/* 执行实际工作：调用回调 */
	fc_ret = fc_ctx->callbacks.filter_prepare_cb(fc_ctx, fc_xid, fc_gid);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;

	return fc_ret;
}

bool filter_by_origin_cb_wrapper(LogicalDecodingContext *fc_ctx, RepOriginId fc_origin_id)
{
	LogicalErrorCallbackState fc_state;
	ErrorContextCallback fc_errcallback;
	bool		fc_ret;

	Assert(!fc_ctx->fast_forward);

	/* 将回调 + 错误上下文信息推入栈中 */
	fc_state.ctx = fc_ctx;
	fc_state.callback_name = "filter_by_origin";
	fc_state.report_location = InvalidXLogRecPtr;
	fc_errcallback.callback = fc_output_plugin_error_callback;
	fc_errcallback.arg = (void *) &fc_state;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 设置输出状态 */
	fc_ctx->accept_writes = false;
	fc_ctx->end_xact = false;

	/* 执行实际工作：调用回调 */
	fc_ret = fc_ctx->callbacks.filter_by_origin_cb(fc_ctx, fc_origin_id);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;

	return fc_ret;
}

static void fc_message_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
				   XLogRecPtr fc_message_lsn, bool fc_transactional,
				   const char *fc_prefix, Size fc_message_size, const char *fc_message)
{
	LogicalDecodingContext *fc_ctx = fc_cache->private_data;
	LogicalErrorCallbackState fc_state;
	ErrorContextCallback fc_errcallback;

	Assert(!fc_ctx->fast_forward);

	if (fc_ctx->callbacks.message_cb == NULL)
		return;

	/* 将回调 + 错误上下文信息推入栈中 */
	fc_state.ctx = fc_ctx;
	fc_state.callback_name = "message";
	fc_state.report_location = fc_message_lsn;
	fc_errcallback.callback = fc_output_plugin_error_callback;
	fc_errcallback.arg = (void *) &fc_state;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 设置输出状态 */
	fc_ctx->accept_writes = true;
	fc_ctx->write_xid = fc_txn != NULL ? fc_txn->xid : InvalidTransactionId;
	fc_ctx->write_location = fc_message_lsn;
	fc_ctx->end_xact = false;

	/* 执行实际工作：调用回调 */
	fc_ctx->callbacks.message_cb(fc_ctx, fc_txn, fc_message_lsn, fc_transactional, fc_prefix,
							  fc_message_size, fc_message);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;
}

static void fc_stream_start_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
						XLogRecPtr fc_first_lsn)
{
	LogicalDecodingContext *fc_ctx = fc_cache->private_data;
	LogicalErrorCallbackState fc_state;
	ErrorContextCallback fc_errcallback;

	Assert(!fc_ctx->fast_forward);

	/* 我们只应该在支持流式处理时调用此函数。 */
	Assert(fc_ctx->streaming);

	/* 将回调 + 错误上下文信息推入栈中 */
	fc_state.ctx = fc_ctx;
	fc_state.callback_name = "stream_start";
	fc_state.report_location = fc_first_lsn;
	fc_errcallback.callback = fc_output_plugin_error_callback;
	fc_errcallback.arg = (void *) &fc_state;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 设置输出状态 */
	fc_ctx->accept_writes = true;
	fc_ctx->write_xid = fc_txn->xid;

	/*
	 * 报告此消息的LSN，以便客户端的回复可以给出最新的
	 * 答复。这永远不会足够（也不应该！）来确认
	 * 收到此事务，但它可能允许确认另一个
	 * 事务的提交只需一条消息。
	 */
	fc_ctx->write_location = fc_first_lsn;

	fc_ctx->end_xact = false;

	/* 在流式模式下，stream_start_cb是必需的 */
	if (fc_ctx->callbacks.stream_start_cb == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("logical streaming requires a %s callback",
						"stream_start_cb")));

	fc_ctx->callbacks.stream_start_cb(fc_ctx, fc_txn);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;
}

static void fc_stream_stop_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
					   XLogRecPtr fc_last_lsn)
{
	LogicalDecodingContext *fc_ctx = fc_cache->private_data;
	LogicalErrorCallbackState fc_state;
	ErrorContextCallback fc_errcallback;

	Assert(!fc_ctx->fast_forward);

	/* 我们只应该在支持流式处理时调用此函数。 */
	Assert(fc_ctx->streaming);

	/* 将回调 + 错误上下文信息推入栈中 */
	fc_state.ctx = fc_ctx;
	fc_state.callback_name = "stream_stop";
	fc_state.report_location = fc_last_lsn;
	fc_errcallback.callback = fc_output_plugin_error_callback;
	fc_errcallback.arg = (void *) &fc_state;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 设置输出状态 */
	fc_ctx->accept_writes = true;
	fc_ctx->write_xid = fc_txn->xid;

	/*
	 * 报告此消息的LSN，以便客户端的回复可以给出最新的
	 * 答复。这永远不会足够（也不应该！）来确认
	 * 收到此事务，但它可能允许确认另一个
	 * 事务的提交只需一条消息。
	 */
	fc_ctx->write_location = fc_last_lsn;

	fc_ctx->end_xact = false;

	/* 在流式模式下，stream_stop_cb是必需的 */
	if (fc_ctx->callbacks.stream_stop_cb == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("logical streaming requires a %s callback",
						"stream_stop_cb")));

	fc_ctx->callbacks.stream_stop_cb(fc_ctx, fc_txn);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;
}

static void fc_stream_abort_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
						XLogRecPtr fc_abort_lsn)
{
	LogicalDecodingContext *fc_ctx = fc_cache->private_data;
	LogicalErrorCallbackState fc_state;
	ErrorContextCallback fc_errcallback;

	Assert(!fc_ctx->fast_forward);

	/* 我们只应该在支持流式处理时调用此函数。 */
	Assert(fc_ctx->streaming);

	/* 将回调 + 错误上下文信息推入栈中 */
	fc_state.ctx = fc_ctx;
	fc_state.callback_name = "stream_abort";
	fc_state.report_location = fc_abort_lsn;
	fc_errcallback.callback = fc_output_plugin_error_callback;
	fc_errcallback.arg = (void *) &fc_state;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 设置输出状态 */
	fc_ctx->accept_writes = true;
	fc_ctx->write_xid = fc_txn->xid;
	fc_ctx->write_location = fc_abort_lsn;
	fc_ctx->end_xact = true;

	/* 在流式模式下，stream_abort_cb是必需的 */
	if (fc_ctx->callbacks.stream_abort_cb == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("logical streaming requires a %s callback",
						"stream_abort_cb")));

	fc_ctx->callbacks.stream_abort_cb(fc_ctx, fc_txn, fc_abort_lsn);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;
}

static void fc_stream_prepare_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
						  XLogRecPtr fc_prepare_lsn)
{
	LogicalDecodingContext *fc_ctx = fc_cache->private_data;
	LogicalErrorCallbackState fc_state;
	ErrorContextCallback fc_errcallback;

	Assert(!fc_ctx->fast_forward);

	/*
	 * 我们只应该在支持流式处理和两阶段提交时调用此函数。
	 */
	Assert(fc_ctx->streaming);
	Assert(fc_ctx->twophase);

	/* 将回调 + 错误上下文信息推入栈中 */
	fc_state.ctx = fc_ctx;
	fc_state.callback_name = "stream_prepare";
	fc_state.report_location = fc_txn->final_lsn;
	fc_errcallback.callback = fc_output_plugin_error_callback;
	fc_errcallback.arg = (void *) &fc_state;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 设置输出状态 */
	fc_ctx->accept_writes = true;
	fc_ctx->write_xid = fc_txn->xid;
	fc_ctx->write_location = fc_txn->end_lsn;
	fc_ctx->end_xact = true;

	/* 在支持两阶段提交的流式模式下，stream_prepare_cb是必需的 */
	if (fc_ctx->callbacks.stream_prepare_cb == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("logical streaming at prepare time requires a %s callback",
						"stream_prepare_cb")));

	fc_ctx->callbacks.stream_prepare_cb(fc_ctx, fc_txn, fc_prepare_lsn);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;
}

static void fc_stream_commit_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
						 XLogRecPtr fc_commit_lsn)
{
	LogicalDecodingContext *fc_ctx = fc_cache->private_data;
	LogicalErrorCallbackState fc_state;
	ErrorContextCallback fc_errcallback;

	Assert(!fc_ctx->fast_forward);

	/* 我们只应该在支持流式处理时调用此函数。 */
	Assert(fc_ctx->streaming);

	/* 将回调 + 错误上下文信息推入栈中 */
	fc_state.ctx = fc_ctx;
	fc_state.callback_name = "stream_commit";
	fc_state.report_location = fc_txn->final_lsn;
	fc_errcallback.callback = fc_output_plugin_error_callback;
	fc_errcallback.arg = (void *) &fc_state;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 设置输出状态 */
	fc_ctx->accept_writes = true;
	fc_ctx->write_xid = fc_txn->xid;
	fc_ctx->write_location = fc_txn->end_lsn;
	fc_ctx->end_xact = true;

	/* 在流式模式下，stream_commit_cb是必需的 */
	if (fc_ctx->callbacks.stream_commit_cb == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("logical streaming requires a %s callback",
						"stream_commit_cb")));

	fc_ctx->callbacks.stream_commit_cb(fc_ctx, fc_txn, fc_commit_lsn);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;
}

static void fc_stream_change_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
						 Relation fc_relation, ReorderBufferChange *fc_change)
{
	LogicalDecodingContext *fc_ctx = fc_cache->private_data;
	LogicalErrorCallbackState fc_state;
	ErrorContextCallback fc_errcallback;

	Assert(!fc_ctx->fast_forward);

	/* 我们只应该在支持流式处理时调用此函数。 */
	Assert(fc_ctx->streaming);

	/* 将回调 + 错误上下文信息推入栈中 */
	fc_state.ctx = fc_ctx;
	fc_state.callback_name = "stream_change";
	fc_state.report_location = fc_change->lsn;
	fc_errcallback.callback = fc_output_plugin_error_callback;
	fc_errcallback.arg = (void *) &fc_state;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 设置输出状态 */
	fc_ctx->accept_writes = true;
	fc_ctx->write_xid = fc_txn->xid;

	/*
	 * 报告此变更的LSN，以便客户端的回复可以给出最新的
	 * 答复。这永远不会足够（也不应该！）来确认
	 * 收到此事务，但它可能允许确认另一个事务的
	 * 提交只需一条消息。
	 */
	fc_ctx->write_location = fc_change->lsn;

	fc_ctx->end_xact = false;

	/* 在流式模式下，stream_change_cb是必需的 */
	if (fc_ctx->callbacks.stream_change_cb == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("logical streaming requires a %s callback",
						"stream_change_cb")));

	fc_ctx->callbacks.stream_change_cb(fc_ctx, fc_txn, fc_relation, fc_change);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;
}

static void fc_stream_message_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
						  XLogRecPtr fc_message_lsn, bool fc_transactional,
						  const char *fc_prefix, Size fc_message_size, const char *fc_message)
{
	LogicalDecodingContext *fc_ctx = fc_cache->private_data;
	LogicalErrorCallbackState fc_state;
	ErrorContextCallback fc_errcallback;

	Assert(!fc_ctx->fast_forward);

	/* 我们只应该在支持流式处理时调用此函数。 */
	Assert(fc_ctx->streaming);

	/* 这个回调是可选的 */
	if (fc_ctx->callbacks.stream_message_cb == NULL)
		return;

	/* 将回调 + 错误上下文信息推入栈中 */
	fc_state.ctx = fc_ctx;
	fc_state.callback_name = "stream_message";
	fc_state.report_location = fc_message_lsn;
	fc_errcallback.callback = fc_output_plugin_error_callback;
	fc_errcallback.arg = (void *) &fc_state;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 设置输出状态 */
	fc_ctx->accept_writes = true;
	fc_ctx->write_xid = fc_txn != NULL ? fc_txn->xid : InvalidTransactionId;
	fc_ctx->write_location = fc_message_lsn;
	fc_ctx->end_xact = false;

	/* 执行实际工作：调用回调 */
	fc_ctx->callbacks.stream_message_cb(fc_ctx, fc_txn, fc_message_lsn, fc_transactional, fc_prefix,
									 fc_message_size, fc_message);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;
}

static void fc_stream_truncate_cb_wrapper(ReorderBuffer *fc_cache, ReorderBufferTXN *fc_txn,
						   int fc_nrelations, Relation fc_relations[],
						   ReorderBufferChange *fc_change)
{
	LogicalDecodingContext *fc_ctx = fc_cache->private_data;
	LogicalErrorCallbackState fc_state;
	ErrorContextCallback fc_errcallback;

	Assert(!fc_ctx->fast_forward);

	/* 我们只应该在支持流式处理时调用此函数。 */
	Assert(fc_ctx->streaming);

	/* 这个回调是可选的 */
	if (!fc_ctx->callbacks.stream_truncate_cb)
		return;

	/* 将回调 + 错误上下文信息推入栈中 */
	fc_state.ctx = fc_ctx;
	fc_state.callback_name = "stream_truncate";
	fc_state.report_location = fc_change->lsn;
	fc_errcallback.callback = fc_output_plugin_error_callback;
	fc_errcallback.arg = (void *) &fc_state;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 设置输出状态 */
	fc_ctx->accept_writes = true;
	fc_ctx->write_xid = fc_txn->xid;

	/*
	 * 报告此变更的LSN，以便客户端的回复可以给出最新的
	 * 答复。这永远不会足够（也不应该！）来确认
	 * 收到此事务，但它可能允许确认另一个事务的
	 * 提交只需一条消息。
	 */
	fc_ctx->write_location = fc_change->lsn;

	fc_ctx->end_xact = false;

	fc_ctx->callbacks.stream_truncate_cb(fc_ctx, fc_txn, fc_nrelations, fc_relations, fc_change);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;
}

/*
 * 为当前复制槽设置所需的目录xmin边界，以获取历史快照。
 *
 * 请注意，在大多数情况下，我们无法立即使用xmin
 * 来增加xmin边界：我们需要等到客户端确认
 * 接收到current_lsn和LogicalConfirmReceivedLocation()后才能继续。
 */
void LogicalIncreaseXminForSlot(XLogRecPtr fc_current_lsn, TransactionId fc_xmin)
{
	bool		fc_updated_xmin = false;
	ReplicationSlot *fc_slot;
	bool		fc_got_new_xmin = false;

	fc_slot = MyReplicationSlot;

	Assert(fc_slot != NULL);

	SpinLockAcquire(&fc_slot->mutex);

	/*
	 * 如果我们已经有了更新的 xmin，则不进行覆盖。这可能发生在我们
	 * 在一个槽中重新启动解码时。
	 */
	if (TransactionIdPrecedesOrEquals(fc_xmin, fc_slot->data.catalog_xmin))
	{
	}

	/*
	 * 如果客户端已经确认到该 lsn，我们可以直接将其标记为已接受。
	 * 这可能发生在我们在一个槽中重新启动解码时。
	 */
	else if (fc_current_lsn <= fc_slot->data.confirmed_flush)
	{
		fc_slot->candidate_catalog_xmin = fc_xmin;
		fc_slot->candidate_xmin_lsn = fc_current_lsn;

		/* 我们的候选者可以直接使用 */
		fc_updated_xmin = true;
	}

	/*
	 * 仅在之前的值已被应用的情况下增加，否则如果接收方确认得太慢，
	 * 我们可能永远无法结束更新。
	 */
	else if (fc_slot->candidate_xmin_lsn == InvalidXLogRecPtr)
	{
		fc_slot->candidate_catalog_xmin = fc_xmin;
		fc_slot->candidate_xmin_lsn = fc_current_lsn;

		/*
		 * 在释放自旋锁之后，以适当的日志级别记录新的 xmin。
		 */
		fc_got_new_xmin = true;
	}
	SpinLockRelease(&fc_slot->mutex);

	if (fc_got_new_xmin)
		elog(DEBUG1, "got new catalog xmin %u at %X/%X", fc_xmin,
			 LSN_FORMAT_ARGS(fc_current_lsn));

	/* 候选者在当前刷新位置下已有效，应用 */
	if (fc_updated_xmin)
		LogicalConfirmReceivedLocation(fc_slot->data.confirmed_flush);
}

/*
 * 标记我们需要读取的最小 LSN (restart_lsn)，以重放在当前 lsn
 * 下未提交的所有事务。
 *
 * 就像 LogicalIncreaseXminForSlot 一样，这只有在客户端确认接收到
 * 当前 lsn 时才生效。
 */
void LogicalIncreaseRestartDecodingForSlot(XLogRecPtr fc_current_lsn, XLogRecPtr fc_restart_lsn)
{
	bool		fc_updated_lsn = false;
	ReplicationSlot *fc_slot;

	fc_slot = MyReplicationSlot;

	Assert(fc_slot != NULL);
	Assert(fc_restart_lsn != InvalidXLogRecPtr);
	Assert(fc_current_lsn != InvalidXLogRecPtr);

	SpinLockAcquire(&fc_slot->mutex);

	/* 如果有更新的重启 lsn，则不进行覆盖 */
	if (fc_restart_lsn <= fc_slot->data.restart_lsn)
	{
		SpinLockRelease(&fc_slot->mutex);
	}

	/*
	 * 我们可能已经刷新得足够远，可以直接接受该 lsn，此时无需
	 * 检查现有候选 LSN。
	 */
	else if (fc_current_lsn <= fc_slot->data.confirmed_flush)
	{
		fc_slot->candidate_restart_valid = fc_current_lsn;
		fc_slot->candidate_restart_lsn = fc_restart_lsn;
		SpinLockRelease(&fc_slot->mutex);

		/* 我们的候选者可以直接使用 */
		fc_updated_lsn = true;
	}

	/*
	 * 仅在之前的值已被应用的情况下增加，否则如果接收方确认得太慢，
	 * 我们可能永远无法结束更新。这里遗漏的值将导致在重新连接后
	 * 需要额外的努力。
	 */
	else if (fc_slot->candidate_restart_valid == InvalidXLogRecPtr)
	{
		fc_slot->candidate_restart_valid = fc_current_lsn;
		fc_slot->candidate_restart_lsn = fc_restart_lsn;
		SpinLockRelease(&fc_slot->mutex);

		elog(DEBUG1, "got new restart lsn %X/%X at %X/%X",
			 LSN_FORMAT_ARGS(fc_restart_lsn),
			 LSN_FORMAT_ARGS(fc_current_lsn));
	}
	else
	{
		XLogRecPtr	fc_candidate_restart_lsn;
		XLogRecPtr	fc_candidate_restart_valid;
		XLogRecPtr	fc_confirmed_flush;

		fc_candidate_restart_lsn = fc_slot->candidate_restart_lsn;
		fc_candidate_restart_valid = fc_slot->candidate_restart_valid;
		fc_confirmed_flush = fc_slot->data.confirmed_flush;
		SpinLockRelease(&fc_slot->mutex);

		elog(DEBUG1, "failed to increase restart lsn: proposed %X/%X, after %X/%X, current candidate %X/%X, current after %X/%X, flushed up to %X/%X",
			 LSN_FORMAT_ARGS(fc_restart_lsn),
			 LSN_FORMAT_ARGS(fc_current_lsn),
			 LSN_FORMAT_ARGS(fc_candidate_restart_lsn),
			 LSN_FORMAT_ARGS(fc_candidate_restart_valid),
			 LSN_FORMAT_ARGS(fc_confirmed_flush));
	}

	/* 候选者在当前刷新位置下已有效，应用 */
	if (fc_updated_lsn)
		LogicalConfirmReceivedLocation(fc_slot->data.confirmed_flush);
}

/*
 * 处理消费者已确认接收到所有更改到 lsn。
 */
void LogicalConfirmReceivedLocation(XLogRecPtr fc_lsn)
{
	Assert(fc_lsn != InvalidXLogRecPtr);

	/* 首先对 candidate_lsn 进行解锁检查。 */
	if (MyReplicationSlot->candidate_xmin_lsn != InvalidXLogRecPtr ||
		MyReplicationSlot->candidate_restart_valid != InvalidXLogRecPtr)
	{
		bool		fc_updated_xmin = false;
		bool		fc_updated_restart = false;
#ifdef FDD
		bool		fc_updated_confirm = false;
#endif

		SpinLockAcquire(&MyReplicationSlot->mutex);

#ifdef FDD
		if (MyReplicationSlot->data.confirmed_flush != fc_lsn)
		{
			fc_updated_confirm = true;
		}
#endif

		MyReplicationSlot->data.confirmed_flush = fc_lsn;

		/* 如果我们已经超出增加 xmin 所需的位置，请执行此操作。 */
		if (MyReplicationSlot->candidate_xmin_lsn != InvalidXLogRecPtr &&
			MyReplicationSlot->candidate_xmin_lsn <= fc_lsn)
		{
			/*
			 * 我们必须在更改内存中的值之前将更改后的 xmin 写入磁盘，
			 * 否则在崩溃后我们将不知道某些目录元组可能已经被
			 * 删除。
			 *
			 * 确保首先写入 ->xmin，并且在新的状态同步到磁盘后
			 * 仅更新 ->effective_xmin。在崩溃后 ->effective_xmin
			 * 被设置为 ->xmin。
			 */
			if (TransactionIdIsValid(MyReplicationSlot->candidate_catalog_xmin) &&
				MyReplicationSlot->data.catalog_xmin != MyReplicationSlot->candidate_catalog_xmin)
			{
				MyReplicationSlot->data.catalog_xmin = MyReplicationSlot->candidate_catalog_xmin;
				MyReplicationSlot->candidate_catalog_xmin = InvalidTransactionId;
				MyReplicationSlot->candidate_xmin_lsn = InvalidXLogRecPtr;
				fc_updated_xmin = true;
			}
		}

		if (MyReplicationSlot->candidate_restart_valid != InvalidXLogRecPtr &&
			MyReplicationSlot->candidate_restart_valid <= fc_lsn)
		{
			Assert(MyReplicationSlot->candidate_restart_lsn != InvalidXLogRecPtr);

			MyReplicationSlot->data.restart_lsn = MyReplicationSlot->candidate_restart_lsn;
			MyReplicationSlot->candidate_restart_lsn = InvalidXLogRecPtr;
			MyReplicationSlot->candidate_restart_valid = InvalidXLogRecPtr;
			fc_updated_restart = true;
		}

		SpinLockRelease(&MyReplicationSlot->mutex);

		/* 首先将新的 xmin 写入磁盘，以便在崩溃后我们知道发生了什么 */
#ifdef FDD
		if (fc_updated_xmin || fc_updated_restart || fc_updated_confirm)
		{
			ReplicationSlotMarkDirty();
			
			if (fc_updated_xmin || fc_updated_restart)
			{				
				ReplicationSlotSave();
				elog(DEBUG1, "updated xmin: %u restart: %u", fc_updated_xmin, fc_updated_restart);
			}
		}
#else
		if (fc_updated_xmin || fc_updated_restart)
		{
			ReplicationSlotMarkDirty();
			ReplicationSlotSave();
			elog(DEBUG1, "updated xmin: %u restart: %u", fc_updated_xmin, fc_updated_restart);
		}
#endif

		/*
		 * 现在新的 xmin 已安全存储在磁盘上，我们可以让全局值
		 * 前进。我们不获取 ProcArrayLock 或其他类似锁，因为我们在
		 * 这里仅前进 xmin，并且并行计算错过这一点不会造成太大伤害。
		 */
		if (fc_updated_xmin)
		{
			SpinLockAcquire(&MyReplicationSlot->mutex);
			MyReplicationSlot->effective_catalog_xmin = MyReplicationSlot->data.catalog_xmin;
			SpinLockRelease(&MyReplicationSlot->mutex);

			ReplicationSlotsComputeRequiredXmin(false);
			ReplicationSlotsComputeRequiredLSN();
		}
	}
	else
	{
#ifndef FDD
        SpinLockAcquire(&MyReplicationSlot->mutex);
		MyReplicationSlot->data.confirmed_flush = fc_lsn;
		SpinLockRelease(&MyReplicationSlot->mutex);
		
#else
		bool fc_dirtied = false;

		SpinLockAcquire(&MyReplicationSlot->mutex);
		if (MyReplicationSlot->data.confirmed_flush != fc_lsn)
		{
			MyReplicationSlot->data.confirmed_flush = fc_lsn;
			fc_dirtied = true;
		}
		SpinLockRelease(&MyReplicationSlot->mutex);

		if (fc_dirtied)
		{
			ReplicationSlotMarkDirty();
		}
#endif
	}
}

/*
 * 在（子）事务回滚期间清除逻辑流状态。
 */
void ResetLogicalStreamingState(void)
{
	CheckXidAlive = InvalidTransactionId;
	bsysscan = false;
}

/*
 * 报告槽的统计信息。
 */
void UpdateDecodingStats(LogicalDecodingContext *fc_ctx)
{
	ReorderBuffer *fc_rb = fc_ctx->reorder;
	PgStat_StatReplSlotEntry fc_repSlotStat;

	/* 如果没有要发送的复制统计信息，则无事可做。 */
	if (fc_rb->spillBytes <= 0 && fc_rb->streamBytes <= 0 && fc_rb->totalBytes <= 0)
		return;

	elog(DEBUG2, "UpdateDecodingStats: updating stats %p %lld %lld %lld %lld %lld %lld %lld %lld",
		 fc_rb,
		 (long long) fc_rb->spillTxns,
		 (long long) fc_rb->spillCount,
		 (long long) fc_rb->spillBytes,
		 (long long) fc_rb->streamTxns,
		 (long long) fc_rb->streamCount,
		 (long long) fc_rb->streamBytes,
		 (long long) fc_rb->totalTxns,
		 (long long) fc_rb->totalBytes);

	fc_repSlotStat.spill_txns = fc_rb->spillTxns;
	fc_repSlotStat.spill_count = fc_rb->spillCount;
	fc_repSlotStat.spill_bytes = fc_rb->spillBytes;
	fc_repSlotStat.stream_txns = fc_rb->streamTxns;
	fc_repSlotStat.stream_count = fc_rb->streamCount;
	fc_repSlotStat.stream_bytes = fc_rb->streamBytes;
	fc_repSlotStat.total_txns = fc_rb->totalTxns;
	fc_repSlotStat.total_bytes = fc_rb->totalBytes;

	pgstat_report_replslot(fc_ctx->slot, &fc_repSlotStat);

	fc_rb->spillTxns = 0;
	fc_rb->spillCount = 0;
	fc_rb->spillBytes = 0;
	fc_rb->streamTxns = 0;
	fc_rb->streamCount = 0;
	fc_rb->streamBytes = 0;
	fc_rb->totalTxns = 0;
	fc_rb->totalBytes = 0;
}
